package com.atguigu.edu.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.beans.TrafficPageViewBean;
import com.atguigu.edu.realtime.utils.BeanList;
import com.atguigu.edu.realtime.utils.DorisUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.util.List;

public class DorisSinkFunction<T> extends RichSinkFunction<T> {
    private Connection jdbcConnection;
    private String tableName;

    public DorisSinkFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
         jdbcConnection = DorisUtil.getJdbcConnection();
    }

    @Override
    public void close() throws Exception {
        DorisUtil.closeJdbcConnection(jdbcConnection);
    }

    @Override
    public void invoke(T orderBean, Context context) throws Exception {

        List<Object> orderList = BeanList.convertObjectToCollection(orderBean);
        //System.out.println("orderList = " + orderList);
        DorisUtil.insertDoris(jdbcConnection,tableName, orderList);

    }
}
